/*
 * Copyright 2017-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.openmessaging.storage.dledger;

import com.alibaba.fastjson.JSON;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import io.openmessaging.storage.dledger.MemberState.Role;
import io.openmessaging.storage.dledger.protocol.DLedgerResponseCode;
import io.openmessaging.storage.dledger.protocol.HeartBeatRequest;
import io.openmessaging.storage.dledger.protocol.HeartBeatResponse;
import io.openmessaging.storage.dledger.protocol.LeadershipTransferRequest;
import io.openmessaging.storage.dledger.protocol.LeadershipTransferResponse;
import io.openmessaging.storage.dledger.protocol.VoteRequest;
import io.openmessaging.storage.dledger.protocol.VoteResponse;
import io.openmessaging.storage.dledger.utils.DLedgerUtils;

/**
 * leader选举
 */
public class DLedgerLeaderElector {
    private static final Logger logger = LoggerFactory.getLogger(DLedgerLeaderElector.class);

    private final Random random = new Random();
    /**
     * 配置
     */
    private final DLedgerConfig dLedgerConfig;
    /**
     * 当前节点状态
     */
    private final MemberState memberState;

    /**
     * DLedger节点之间发送rpc请求的服务.
     */
    private final DLedgerRpcService dLedgerRpcService;

    // =====================================  as a server handler =========================================

    //record the last leader state
    private volatile long lastLeaderHeartBeatTime = -1;
    private volatile long lastSendHeartBeatTime = -1;
    private volatile long lastSuccHeartBeatTime = -1;

    /**
     * 发送心跳的时间间隔 默认每隔2秒发送一次心跳
     */
    private int heartBeatTimeIntervalMs = 2000;
    /**
     * 最大心跳泄漏次数
     *
     * 解释:从上一次成功心跳时间开始,Leader给追随者发送了3次心跳都没有追随者'回应'(追随者不说成功,也不说失败,就和Leader玩失踪,不搭理Leader),
     *      则说明当前Leader已经出现分区(Leader联系不上自己的追随者了.)
     */
    private int maxHeartBeatLeak = 3;


    // =====================================  as a client =========================================

    /**
     * 下一次发送投票请求的时间
     */
    private long nextTimeToRequestVote = -1;
    /**
     * 投票时,是否立即增加任期
     *
     * 注意:这个变量是在运行期会不断变化,并不是启动配置.
     */
    private volatile boolean needIncreaseTermImmediately = false;
    private int minVoteIntervalMs = 300;
    private int maxVoteIntervalMs = 1000;

    /**
     * 角色变化处理器
     */
    private List<RoleChangeHandler> roleChangeHandlers = new ArrayList<>();

    /**
     * 最后一次投票的结果
     */
    private VoteResponse.ParseResult lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;

    /**
     * 角色维护线程
     */
    private final StateMaintainer stateMaintainer = new StateMaintainer("StateMaintainer", logger);

    private final TakeLeadershipTask takeLeadershipTask = new TakeLeadershipTask();

    public DLedgerLeaderElector(DLedgerConfig dLedgerConfig, MemberState memberState,
                                DLedgerRpcService dLedgerRpcService) {
        this.dLedgerConfig = dLedgerConfig;
        this.memberState = memberState;
        this.dLedgerRpcService = dLedgerRpcService;
        refreshIntervals(dLedgerConfig);
    }

    public void startup() {
        // 启动线程
        stateMaintainer.start();
        for (RoleChangeHandler roleChangeHandler : roleChangeHandlers) {
            roleChangeHandler.startup();
        }
    }

    public void shutdown() {
        stateMaintainer.shutdown();
        for (RoleChangeHandler roleChangeHandler : roleChangeHandlers) {
            roleChangeHandler.shutdown();
        }
    }

    /**
     * 重新设置各种间隔
     *
     * @param dLedgerConfig 配置
     */
    private void refreshIntervals(DLedgerConfig dLedgerConfig) {
        this.heartBeatTimeIntervalMs = dLedgerConfig.getHeartBeatTimeIntervalMs();
        this.maxHeartBeatLeak = dLedgerConfig.getMaxHeartBeatLeak();
        this.minVoteIntervalMs = dLedgerConfig.getMinVoteIntervalMs();
        this.maxVoteIntervalMs = dLedgerConfig.getMaxVoteIntervalMs();
    }

    /**
     * Follower处理心跳请求.
     *
     * @param request Leader给Follower发的心跳请求
     */
    public CompletableFuture<HeartBeatResponse> handleHeartBeat(HeartBeatRequest request) throws Exception {

        if (!memberState.isPeerMember(request.getLeaderId())) {
            logger.warn("[BUG] [HandleHeartBeat] remoteId={} is an unknown member", request.getLeaderId());
            return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.UNKNOWN_MEMBER.getCode()));
        }

        // 处理心跳的节点一定是Follower节点.如果当前节点是Leader,就需要返回错误.
        if (memberState.getSelfId().equals(request.getLeaderId())) {
            logger.warn("[BUG] [HandleHeartBeat] selfId={} but remoteId={}", memberState.getSelfId(), request.getLeaderId());
            return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.UNEXPECTED_MEMBER.getCode()));
        }

        if (request.getTerm() < memberState.currTerm()) {
            return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
        } else if (request.getTerm() == memberState.currTerm()) {
            if (request.getLeaderId().equals(memberState.getLeaderId())) {
                lastLeaderHeartBeatTime = System.currentTimeMillis();
                // 心跳的term相同 leaderId相同. 返回成功.
                return CompletableFuture.completedFuture(HeartBeatResponse.success());
            }
        }

        //abnormal case    异常情况:
        //hold the lock to get the latest term and leaderId
        // 获取锁,然后获取最新的term和leaderId再去判断一下.
        synchronized (memberState) {
            if (request.getTerm() < memberState.currTerm()) {
                return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
            } else if (request.getTerm() == memberState.currTerm()) {
                if (memberState.getLeaderId() == null) {
                    // Leader和当前这个Follower的term相同,但是Follower还没有leaderId.则设置leaderId
                    changeRoleToFollower(request.getTerm(), request.getLeaderId());
                    return CompletableFuture.completedFuture(HeartBeatResponse.success());
                } else if (request.getLeaderId().equals(memberState.getLeaderId())) {
                    lastLeaderHeartBeatTime = System.currentTimeMillis();
                    return CompletableFuture.completedFuture(HeartBeatResponse.success());
                } else {
                    //this should not happen, but if happened
                    // 这不应该发生，但如果发生了.则抛出异常.
                    logger.error("[{}][BUG] currTerm {} has leader {}, but received leader {}", memberState.getSelfId(), memberState.currTerm(), memberState.getLeaderId(), request.getLeaderId());
                    return CompletableFuture.completedFuture(new HeartBeatResponse().code(DLedgerResponseCode.INCONSISTENT_LEADER.getCode()));
                }
            } else { // Leader term > Follower term 的情况.
                //To make it simple, for larger term, do not change to follower immediately, first change to candidate, and notify the state-maintainer thread
                // 为了简单起见，对于更大的任期，不要立即更改为follower，先更改为candidate，并通知state-maintainer线程
                changeRoleToCandidate(request.getTerm());
                // 标记:立即增加term
                needIncreaseTermImmediately = true;
                //TOOD notify
                return CompletableFuture.completedFuture(new HeartBeatResponse().code(DLedgerResponseCode.TERM_NOT_READY.getCode()));
            }
        }
    }

    public void changeRoleToLeader(long term) {
        synchronized (memberState) {
            if (memberState.currTerm() == term) {
                memberState.changeToLeader(term);
                lastSendHeartBeatTime = -1;
                handleRoleChange(term, Role.LEADER);
                logger.info("[{}] [ChangeRoleToLeader] from term: {} and currTerm: {}", memberState.getSelfId(), term, memberState.currTerm());
            } else {
                logger.warn("[{}] skip to be the leader in term: {}, but currTerm is: {}", memberState.getSelfId(), term, memberState.currTerm());
            }
        }
    }

    /**
     * 修改当前角色为 {@link Role#CANDIDATE}
     *
     * 不会修改当前节点的 term 任期.这里只是把角色改成 {@link Role#CANDIDATE}
     */
    public void changeRoleToCandidate(long term) {
        synchronized (memberState) {
            if (term >= memberState.currTerm()) {
                memberState.changeToCandidate(term);
                handleRoleChange(term, Role.CANDIDATE);
                logger.info("[{}] [ChangeRoleToCandidate] from term: {} and currTerm: {}", memberState.getSelfId(), term, memberState.currTerm());
            } else {
                logger.info("[{}] skip to be candidate in term: {}, but currTerm: {}", memberState.getSelfId(), term, memberState.currTerm());
            }
        }
    }

    //just for test
    public void testRevote(long term) {
        changeRoleToCandidate(term);
        lastParseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
        nextTimeToRequestVote = -1;
    }

    public void changeRoleToFollower(long term, String leaderId) {
        logger.info("[{}][ChangeRoleToFollower] from term: {} leaderId: {} and currTerm: {}", memberState.getSelfId(), term, leaderId, memberState.currTerm());
        lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
        memberState.changeToFollower(term, leaderId);
        lastLeaderHeartBeatTime = System.currentTimeMillis();
        handleRoleChange(term, Role.FOLLOWER);
    }

    /**
     * 处理投票请求
     *
     * @param request 投票请求
     * @param self    投票是否投给自己
     */
    public CompletableFuture<VoteResponse> handleVote(VoteRequest request, boolean self) {
        //hold the lock to get the latest term, leaderId, ledgerEndIndex
        // 使用锁,避免并发: 对比对方和自己的term、leaderId、ledgerEndIndex的值.
        synchronized (memberState) {
            if (!memberState.isPeerMember(request.getLeaderId())) {
                logger.warn("[BUG] [HandleVote] remoteId={} is an unknown member", request.getLeaderId());
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_UNKNOWN_LEADER));
            }

            // 如果不是自己投票给自己,那么selfId和req.leaderId不可能相等.
            if (!self && memberState.getSelfId().equals(request.getLeaderId())) {
                logger.warn("[BUG] [HandleVote] selfId={} but remoteId={}", memberState.getSelfId(), request.getLeaderId());
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_UNEXPECTED_LEADER));
            }

            // 说明:对方的index需要>=自己的;并且 term需要相等,才会同意对方.
            if (request.getLedgerEndTerm() < memberState.getLedgerEndTerm()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_EXPIRED_LEDGER_TERM));
            } else if (request.getLedgerEndTerm() == memberState.getLedgerEndTerm() && request.getLedgerEndIndex() < memberState.getLedgerEndIndex()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_SMALL_LEDGER_END_INDEX));
            }

            if (request.getTerm() < memberState.currTerm()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_EXPIRED_VOTE_TERM));
            }

            // 相同的投票任期
            //  只有term任期相同的以下两种情况才会继续往下走:
            //  - 1、当前节点还没给任何人投票.(可能是自己给自己投票,也可能是先接收到了对方的投票请求.)
            //  - 2、对方和自己投票给相同的候选人.
            else if (request.getTerm() == memberState.currTerm()) {
                if (memberState.currVoteFor() == null) {
                    // note:1、当前节点还没给任何人投票.(可能是自己给自己投票,也可能是先接收到了对方的投票请求.)
                } else if (memberState.currVoteFor().equals(request.getLeaderId())) {
                    // note:2、对方和自己投票给相同的候选人.        repeat just let it go
                } else {
                    // 对方和自己投票给不同的候选人 并且 当前节点已存在LeaderId——已经选举出Leader节点了.
                    if (memberState.getLeaderId() != null) {
                        return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_ALREADY_HAS_LEADER));
                    } else {
                        // 对方和自己投票给不同的候选人 但是尚未选出Leader.   (举例:当前是节点A收到B的投票消息,A和B投票给不同的候选人)
                        return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_ALREADY_VOTED));
                    }
                }
            }

            // req.term > currTerm    说明:对方开启了新一轮投票
            else {
                // 被更大的任期降级,角色改成 Candidate       stepped down by larger term
                changeRoleToCandidate(request.getTerm());    // 当前角色改成 Candidate ,并使用对方的term
                needIncreaseTermImmediately = true;          // 对方开启了一个新的任期,所以当前节点,需要立即增加 currTerm
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_TERM_NOT_READY));
            }

            if (request.getTerm() < memberState.getLedgerEndTerm()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.getLedgerEndTerm()).voteResult(VoteResponse.RESULT.REJECT_TERM_SMALL_THAN_LEDGER));
            }

            // 不是自己给自己投票 并且 当前节点是首选Leader
            if (!self && isTakingLeadership() && request.getLedgerEndTerm() == memberState.getLedgerEndTerm() && memberState.getLedgerEndIndex() >= request.getLedgerEndIndex()) {
                return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.REJECT_TAKING_LEADERSHIP));
            }

            // todo:疑问:如果term、index都相等,会使用什么逻辑判断,投票给谁呢?
            //  解答:举个例子:集群中有3个节点,投票是term和index相等,并且每个节点都投票给自己(这种情况就很极端,但是存在发生的可能性).
            //       投票统计时,就会把状态设置成 WAIT_TO_VOTE_NEXT ,当线程下一次执行时,就会获取下一个term.然后再次去发起投票.

            // 当前节点投票给谁.
            // 有以下两种情况会修改投票id:
            //  - 第一次投票时,自己给自己投票
            //  - 当前候选人A收到B的投票,A认同B的投票时:
            //    - term相同、ledgerEndTerm相同、req.getLedgerEndIndex() > memberState.getLedgerEndIndex()
            memberState.setCurrVoteFor(request.getLeaderId());
            return CompletableFuture.completedFuture(new VoteResponse(request).term(memberState.currTerm()).voteResult(VoteResponse.RESULT.ACCEPT));
        }
    }

    /**
     * {@link Role#LEADER} 给{@link Role#FOLLOWER}和{@link Role#CANDIDATE}发送心跳请求
     * 
     * 
     * Leader节点给其他节点发送心跳,其他节点收到心跳request之后,做出response:
     * - 处理成功 (认同Leader领导)                                   {@link DLedgerResponseCode#SUCCESS}
     * - 开启下一个任期选举                                          {@link DLedgerResponseCode#EXPIRED_TERM}
     * - 不一致的领导者(Follower选出新的Leader)                      {@link DLedgerResponseCode#INCONSISTENT_LEADER}
     * - 任期未准备好(认同Leader领导,但是还需要做一同步工作)           {@link DLedgerResponseCode#TERM_NOT_READY}
     *
     * @param term     任期
     * @param leaderId Leader节点的id
     */
    private void sendHeartbeats(long term, String leaderId) throws Exception {
        final AtomicInteger allNum = new AtomicInteger(1);
        // 处理心跳请求成功的Follower数量
        final AtomicInteger succNum = new AtomicInteger(1);
        final AtomicInteger notReadyNum = new AtomicInteger(0);
        final AtomicLong maxTerm = new AtomicLong(-1);
        final AtomicBoolean inconsistLeader = new AtomicBoolean(false);
        final CountDownLatch beatLatch = new CountDownLatch(1);
        long startHeartbeatTimeMs = System.currentTimeMillis();
        // 遍历集群中所有节点
        for (String peerId : memberState.getPeerMap().keySet()) {
            // 不给自己发送心跳数据
            if (memberState.getSelfId().equals(peerId)) {
                continue;
            }
            // 构造心跳请求
            HeartBeatRequest heartBeatRequest = new HeartBeatRequest();
            heartBeatRequest.setGroup(memberState.getGroup());     // 集群组名
            heartBeatRequest.setLocalId(memberState.getSelfId());  // 当前节点自己的id
            heartBeatRequest.setRemoteId(peerId);                      // 对方的id
            heartBeatRequest.setLeaderId(leaderId);                // Leader节点的id
            heartBeatRequest.setTerm(term);                        // 任期

            CompletableFuture<HeartBeatResponse> future = dLedgerRpcService.heartBeat(heartBeatRequest);
            future.whenComplete((HeartBeatResponse x, Throwable ex) -> {
                try {
                    if (ex != null) { // 异步心跳发送异常
                        // 设置对方存活状态为false
                        memberState.getPeersLiveTable().put(peerId, Boolean.FALSE);
                        // 抛出异常
                        throw ex;
                    }
                    switch (DLedgerResponseCode.valueOf(x.getCode())) {
                        case SUCCESS: // Follower处理心跳请求成功:增加成功的数量
                            succNum.incrementAndGet();
                            break;
                        case EXPIRED_TERM:
                            maxTerm.set(x.getTerm());
                            break;
                        case INCONSISTENT_LEADER: // 不一致的领导者 (Follower觉得LeaderId不一样)   官方注释:this should not happen, but if happened
                            inconsistLeader.compareAndSet(false, true);
                            break;
                        case TERM_NOT_READY: // 任期未准备好:  Leader的term > Follower的term    的情况.
                            notReadyNum.incrementAndGet();
                            break;
                        default:
                            break;
                    }

                    if (x.getCode() == DLedgerResponseCode.NETWORK_ERROR.getCode()) {
                        memberState.getPeersLiveTable().put(peerId, Boolean.FALSE);
                    } else {
                        // 记录Follower是存活状态的.
                        memberState.getPeersLiveTable().put(peerId, Boolean.TRUE);
                    }

                    // 法定人数的Follower心跳成功.
                    if (memberState.isQuorum(succNum.get())
                        || memberState.isQuorum(succNum.get() + notReadyNum.get())) {
                        beatLatch.countDown();
                    }
                } catch (Throwable t) {
                    logger.error("heartbeat response failed", t);
                } finally {
                    allNum.incrementAndGet();
                    // 所有心跳请求都处理完毕后,就无需等待了.
                    if (allNum.get() == memberState.peerSize()) {
                        beatLatch.countDown();
                    }
                }
            });
        } // for结束:给所有peer异步发送完心跳请求

        // 默认等待2秒.(心跳间隔默认2秒)   有两种情况会提前结束等待:
        //  1、处理成功的节点,超过半数.
        //  2、所有心跳请求都响应完毕.
        beatLatch.await(heartBeatTimeIntervalMs, TimeUnit.MILLISECONDS);


        // ====================== 超过半数Follower成功. ======================
        if (memberState.isQuorum(succNum.get())) {
            // 设置最后一次成功的时间
            lastSuccHeartBeatTime = System.currentTimeMillis();
            return;
        }


        // ====================== 没有法定人数的Follower成功处理心跳 ======================
        logger.info("[{}] Parse heartbeat responses in cost={} term={} allNum={} succNum={} notReadyNum={} inconsistLeader={} maxTerm={} peerSize={} lastSuccHeartBeatTime={}",
                    memberState.getSelfId(), DLedgerUtils.elapsed(startHeartbeatTimeMs), term, allNum.get(), succNum.get(), notReadyNum.get(), inconsistLeader.get(), maxTerm.get(), memberState.peerSize(), new Timestamp(lastSuccHeartBeatTime));

        // 成功的Follower和未准备好的Follower,已超过半数.
        if (memberState.isQuorum(succNum.get() + notReadyNum.get())) {
            // 把最后一次发送心跳时间设置成-1.目的:让下一次心跳请求立即发送(不需要等待2秒再发送下一次心跳)
            lastSendHeartBeatTime = -1;
        }
        // 最大任期大于当前Leader的任期(说明Follower开启了新一轮选主,但是尚未选出新的Leader)
        else if (maxTerm.get() > term) {
            changeRoleToCandidate(maxTerm.get());
        }
        // 不一样的Leader(追随者已经有了新老大)
        else if (inconsistLeader.get()) {
            changeRoleToCandidate(term);
        }
        // Leader发送3此心跳,追随者都没有回应(老大联系不上自己的追随者了).
        else if (DLedgerUtils.elapsed(lastSuccHeartBeatTime) > maxHeartBeatLeak * heartBeatTimeIntervalMs) {
            changeRoleToCandidate(term);
        }
    }

    /**
     * 保持为Leader
     */
    private void maintainAsLeader() throws Exception {
        // 默认每隔2秒发送一次心跳包,这里根据最后一次心跳时间判断是否应该发送心跳了.
        if (DLedgerUtils.elapsed(lastSendHeartBeatTime) > heartBeatTimeIntervalMs) {
            long term;
            String leaderId;
            synchronized (memberState) {
                // 如果当前节点不是Leader,则不做任何事.
                if (!memberState.isLeader()) {
                    //stop sending
                    return;
                }
                // 发送心跳包之前,修改数据.
                term = memberState.currTerm();
                leaderId = memberState.getLeaderId();
                lastSendHeartBeatTime = System.currentTimeMillis();
            }
            // 发送心跳
            sendHeartbeats(term, leaderId);
        }
    }

    private void maintainAsFollower() {
        if (DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > 2 * heartBeatTimeIntervalMs) {
            synchronized (memberState) {
                // 当前角色是Follower,并且已经超过3次没有收到Leader的心跳请求,则修改角色 Follower -> Candidate
                if (memberState.isFollower() && (DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > maxHeartBeatLeak * heartBeatTimeIntervalMs)) {
                    logger.info("[{}][HeartBeatTimeOut] lastLeaderHeartBeatTime: {} heartBeatTimeIntervalMs: {} lastLeader={}", memberState.getSelfId(), new Timestamp(lastLeaderHeartBeatTime), heartBeatTimeIntervalMs, memberState.getLeaderId());
                    changeRoleToCandidate(memberState.currTerm());
                }
            }
        }
    }

    /**
     * 给所有的成员发送投票请求.
     * 1.如果peerId是自己的话,就直接调用 {@link DLedgerLeaderElector#handleVote(VoteRequest, boolean)} 方法.
     * 2.如果peerId不是自己,则发送rpc请求.
     *
     * @param term
     * @param ledgerEndTerm
     * @param ledgerEndIndex
     * @return
     * @throws Exception
     */
    private List<CompletableFuture<VoteResponse>> voteForQuorumResponses(long term, long ledgerEndTerm, long ledgerEndIndex) throws Exception {
        List<CompletableFuture<VoteResponse>> responses = new ArrayList<>();
        for (String peerId : memberState.getPeerMap().keySet()) {
            VoteRequest voteRequest = new VoteRequest();       // 投票请求javaBean
            voteRequest.setGroup(memberState.getGroup());      // groupName
            voteRequest.setLedgerEndIndex(ledgerEndIndex);     // 当前节点最后的下标 (类似ZooKeeper的事务id)
            voteRequest.setLedgerEndTerm(ledgerEndTerm);       // 任期
            voteRequest.setLeaderId(memberState.getSelfId());  // 每个节点的第一票都是投票给自己
            voteRequest.setTerm(term);                         // todo:term和ledgerEndTerm区别是?
            voteRequest.setRemoteId(peerId);                   // 远程节点id
            if (memberState.getSelfId().equals(peerId)) {
                // 当前节点,第一票先自己投票给自己
                responses.add(handleVote(voteRequest, true));
            } else {
                //async 异步发送投票请求
                responses.add(dLedgerRpcService.vote(voteRequest));
            }
        }
        return responses;
    }

    /**
     * 判断当前节点是否是首选Leader
     *
     * @return 当前节点是首选Leader服务器, 返回true
     * @see DLedgerConfig#getPreferredLeaderIds()
     */
    private boolean isTakingLeadership() {
        // 当前节点是首选Leader   并且    当前节点是受让人
        if (dLedgerConfig.getPreferredLeaderIds() != null && memberState.getTermToTakeLeadership() == memberState.currTerm()) {
            List<String> preferredLeaderIds = Arrays.asList(dLedgerConfig.getPreferredLeaderIds().split(";"));
            return preferredLeaderIds.contains(memberState.getSelfId());
        }
        return false;
    }

    /**
     * 获取下一发送投票请求的时间
     */
    private long getNextTimeToRequestVote() {
        if (isTakingLeadership()) {
            return System.currentTimeMillis() + dLedgerConfig.getMinTakeLeadershipVoteIntervalMs() +
                random.nextInt(dLedgerConfig.getMaxTakeLeadershipVoteIntervalMs() - dLedgerConfig.getMinTakeLeadershipVoteIntervalMs());
        }
        return System.currentTimeMillis() + minVoteIntervalMs + random.nextInt(maxVoteIntervalMs - minVoteIntervalMs);
    }

    /**
     * 当前节点是 {@link Role#CANDIDATE} ,开始选举投票
     *
     * 1、当前时间尚未到达下次投票时间 并且 needIncreaseTermImmediately=false(不需要立即增加term) 时,方法直接return,即:直接跳过本次投票.
     */
    private void maintainAsCandidate() throws Exception {
        //for candidate
        if (System.currentTimeMillis() < nextTimeToRequestVote && !needIncreaseTermImmediately) {
            return;
        }
        // 投票任期
        long term;
        long ledgerEndTerm;
        long ledgerEndIndex;
        synchronized (memberState) {
            if (!memberState.isCandidate()) { // 不是Candidate角色,就不投票
                return;
            }
            // 这个判断为了:确定是否需要立即增加 term
            if (lastParseResult == VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT || needIncreaseTermImmediately) {
                long prevTerm = memberState.currTerm();
                // needIncreaseTermImmediately == true 时, ++currTerm  表示:当前节点的当前term自增.(开启新一轮投票vote)
                // 获取下一个投票任期
                term = memberState.nextTerm();
                logger.info("{}_[INCREASE_TERM] from {} to {}", memberState.getSelfId(), prevTerm, term);
                lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
            } else { // 不需要增加投票term的情况.
                // 获取当前投票任期
                term = memberState.currTerm();
            }
            ledgerEndIndex = memberState.getLedgerEndIndex();
            ledgerEndTerm = memberState.getLedgerEndTerm();
        }
        // 是否立即增加任期
        if (needIncreaseTermImmediately) {
            // 获取下一发送投票请求的时间
            nextTimeToRequestVote = getNextTimeToRequestVote();
            needIncreaseTermImmediately = false; // 增加完任期之后,设置为false.(下一次线程的任务执行时,就不再增加任期term了.)
            return;
        }

        // 开始给所有peer发送投票请求
        long startVoteTimeMs = System.currentTimeMillis();
        final List<CompletableFuture<VoteResponse>> quorumVoteResponses = voteForQuorumResponses(term, ledgerEndTerm, ledgerEndIndex);
        final AtomicLong knownMaxTermInGroup = new AtomicLong(term);
        final AtomicInteger allNum = new AtomicInteger(0);
        final AtomicInteger validNum = new AtomicInteger(0);
        // 同意自己投票的候选人数量
        final AtomicInteger acceptedNum = new AtomicInteger(0);
        final AtomicInteger notReadyTermNum = new AtomicInteger(0);
        final AtomicInteger biggerLedgerNum = new AtomicInteger(0);
        // 已选出Leader
        final AtomicBoolean alreadyHasLeader = new AtomicBoolean(false);

        // 处理所有peer返回的投票response
        CountDownLatch voteLatch = new CountDownLatch(1);
        for (CompletableFuture<VoteResponse> future : quorumVoteResponses) {
            future.whenComplete((VoteResponse resp, Throwable ex) -> {
                try {
                    if (ex != null) {
                        throw ex;
                    }
                    logger.info("[{}][GetVoteResponse] {}", memberState.getSelfId(), JSON.toJSONString(resp));
                    if (resp.getVoteResult() != VoteResponse.RESULT.UNKNOWN) {
                        validNum.incrementAndGet();
                    }
                    synchronized (knownMaxTermInGroup) {
                        switch (resp.getVoteResult()) {
                            case ACCEPT: // 对方候选人同意自己的投票
                                acceptedNum.incrementAndGet();
                                break;
                            case REJECT_ALREADY_VOTED:
                            case REJECT_TAKING_LEADERSHIP:
                                break;
                            case REJECT_ALREADY_HAS_LEADER:  // Leader已经选出
                                alreadyHasLeader.compareAndSet(false, true);
                                break;
                            case REJECT_TERM_SMALL_THAN_LEDGER:
                            case REJECT_EXPIRED_VOTE_TERM:   // 自己的term < 对方的term
                                if (resp.getTerm() > knownMaxTermInGroup.get()) {
                                    knownMaxTermInGroup.set(resp.getTerm());
                                }
                                break;
                            case REJECT_EXPIRED_LEDGER_TERM:
                            case REJECT_SMALL_LEDGER_END_INDEX:
                                biggerLedgerNum.incrementAndGet();
                                break;
                            case REJECT_TERM_NOT_READY:
                                notReadyTermNum.incrementAndGet();
                                break;
                            default:
                                break;
                        }
                    }
                    // 如果符合任意一个条件,不等待
                    if (alreadyHasLeader.get()                       // 已存在Leader
                        || memberState.isQuorum(acceptedNum.get())   // 超过半数的候选人,同意自己的投票
                        || memberState.isQuorum(acceptedNum.get() + notReadyTermNum.get())) {  // 超过半数:同意自己的候选人 和 req.term > resp.term
                        voteLatch.countDown();
                    }
                } catch (Throwable t) {
                    logger.error("vote response failed", t);
                } finally {
                    allNum.incrementAndGet();
                    if (allNum.get() == memberState.peerSize()) {
                        voteLatch.countDown();
                    }
                }
            });
        } // for结束:投票future处理完毕.

        try {
            // 最多等待3秒钟
            voteLatch.await(2000 + random.nextInt(maxVoteIntervalMs), TimeUnit.MILLISECONDS);
        } catch (Throwable ignore) {

        }

        // 投票花费的时间
        long lastVoteCost = DLedgerUtils.elapsed(startVoteTimeMs);

        VoteResponse.ParseResult parseResult;
        // 已知最大的term大于当前自己的term,说明,别人已经开启了一轮新投票
        if (knownMaxTermInGroup.get() > term) {
            parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
            nextTimeToRequestVote = getNextTimeToRequestVote();
            changeRoleToCandidate(knownMaxTermInGroup.get());
        }
        // 单个候选人说已经选出Leader,那指定是存在'叛徒'.
        // 解释:投票发给其他候选人时,候选人如果认同会返回 ACCEPT ,如果超过半数同意,才会选出Leader角色.
        //      所有如果某个候选人说已经选出Leader了,则需要等待下一次投票.
        else if (alreadyHasLeader.get()) {
            parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
            nextTimeToRequestVote = getNextTimeToRequestVote() + heartBeatTimeIntervalMs * maxHeartBeatLeak;
        }
        // 没有超过半数:response 为null的请求,没有超过半数
        else if (!memberState.isQuorum(validNum.get())) {
            // 重新投票
            parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
            nextTimeToRequestVote = getNextTimeToRequestVote();
        }
        else if (!memberState.isQuorum(validNum.get() - biggerLedgerNum.get())) {
            parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
            nextTimeToRequestVote = getNextTimeToRequestVote() + maxVoteIntervalMs;
        }
        // note:[同意自己投票的候选人数量,超过半数,则说明投票通过]
        //  DLedger投票和ZooKeeper投票不一样:
        //  - ZooKeeper投票:如果A认同B的投票,A会把B的投票发送出去.
        //  - DLedger投票:A给B、C发送投票,如果B、C同意,则超过半数.则A就是leader.
        else if (memberState.isQuorum(acceptedNum.get())) {
            parseResult = VoteResponse.ParseResult.PASSED;
        }
        // 同意自己和未准备好的候选人数量,超过半数 —— 立即重新投票
        else if (memberState.isQuorum(acceptedNum.get() + notReadyTermNum.get())) {
            parseResult = VoteResponse.ParseResult.REVOTE_IMMEDIATELY;
        }
        else {
            // 等待下一次投票
            parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
            // 设置下一发送投票请求的时间
            nextTimeToRequestVote = getNextTimeToRequestVote();
        }
        lastParseResult = parseResult;
        logger.info("[{}] [PARSE_VOTE_RESULT] cost={} term={} memberNum={} allNum={} acceptedNum={} notReadyTermNum={} biggerLedgerNum={} alreadyHasLeader={} maxTerm={} result={}", memberState.getSelfId(), lastVoteCost, term, memberState.peerSize(), allNum, acceptedNum, notReadyTermNum, biggerLedgerNum, alreadyHasLeader, knownMaxTermInGroup.get(), parseResult);

        // 如果投票[PASSED],则当前节点变成Leader角色
        if (parseResult == VoteResponse.ParseResult.PASSED) {
            logger.info("[{}] [VOTE_RESULT] has been elected to be the leader in term {}", memberState.getSelfId(), term);
            changeRoleToLeader(term);
        }
    }

    /**
     * The core method of maintainer. Run the specified logic according to the current role: candidate => propose a
     * vote. leader => send heartbeats to followers, and step down to candidate when quorum followers do not respond.
     * follower => accept heartbeats, and change to candidate when no heartbeat from leader.
     *
     * 角色维护的核心方法.
     * 根据当前角色运行指定逻辑：
     * {@link Role#CANDIDATE} =>  提出投票。
     * {@link Role#LEADER}    =>  向追随者发送心跳，并在法定(超过半数的)追随者没有响应时降级为候选人。
     * {@link Role#FOLLOWER}  =>  接受心跳，当没有leader心跳时切换为候选。
     */
    private void maintainState() throws Exception {
        if (memberState.isLeader()) {
            maintainAsLeader();
        } else if (memberState.isFollower()) {
            maintainAsFollower();
        } else {
            maintainAsCandidate();
        }
    }

    /**
     * 处理角色变化
     *
     * @param term    当前任期
     * @param newRole 新角色
     */
    private void handleRoleChange(long term, Role newRole) {
        try {
            // 检查 Leader转让的逻辑.如果是 受让人,才会真正去校验
            takeLeadershipTask.check(term, newRole);
        } catch (Throwable t) {
            logger.error("takeLeadershipTask.check failed. ter={}, role={}", term, newRole, t);
        }

        for (RoleChangeHandler roleChangeHandler : roleChangeHandlers) {
            try {
                roleChangeHandler.handle(term, newRole);
            } catch (Throwable t) {
                logger.warn("Handle role change failed term={} role={} handler={}", term, newRole, roleChangeHandler.getClass(), t);
            }
        }
    }

    public void addRoleChangeHandler(RoleChangeHandler roleChangeHandler) {
        if (!roleChangeHandlers.contains(roleChangeHandler)) {
            roleChangeHandlers.add(roleChangeHandler);
        }
    }

    /**
     * Leader节点转换给其他节点.
     *
     * 如果当前节点是Leader角色,则此节点可以把自己的Leader角色转移给别的节点.
     */
    public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(LeadershipTransferRequest request) throws Exception {
        logger.info("handleLeadershipTransfer: {}", request);
        synchronized (memberState) {
            if (memberState.currTerm() != request.getTerm()) {
                logger.warn("[BUG] [HandleLeaderTransfer] currTerm={} != request.term={}", memberState.currTerm(), request.getTerm());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.INCONSISTENT_TERM.getCode()));
            }

            // 当前节点不是Leader,则返回.
            if (!memberState.isLeader()) {
                logger.warn("[BUG] [HandleLeaderTransfer] selfId={} is not leader", request.getLeaderId());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.NOT_LEADER.getCode()));
            }

            // 如果已经有Leader转换信息了,则说明当前正在转换中.等待完成就行了.
            if (memberState.getTransferee() != null) {
                logger.warn("[BUG] [HandleLeaderTransfer] transferee={} is already set", memberState.getTransferee());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.LEADER_TRANSFERRING.getCode()));
            }

            // 当前是Leader节点,设置受让人id
            memberState.setTransferee(request.getTransfereeId());
        }
        LeadershipTransferRequest takeLeadershipRequest = new LeadershipTransferRequest();
        takeLeadershipRequest.setGroup(memberState.getGroup());
        takeLeadershipRequest.setLeaderId(memberState.getLeaderId());
        takeLeadershipRequest.setLocalId(memberState.getSelfId());
        takeLeadershipRequest.setRemoteId(request.getTransfereeId()); // 远程id:给受让人发送消息
        takeLeadershipRequest.setTerm(request.getTerm());
        takeLeadershipRequest.setTakeLeadershipLedgerIndex(memberState.getLedgerEndIndex()); // 当前Leader节点最大的index
        takeLeadershipRequest.setTransferId(memberState.getSelfId());                        // 转让人id(也就是当前Leader节点的id)
        takeLeadershipRequest.setTransfereeId(request.getTransfereeId());                    // 受让人id

        if (memberState.currTerm() != request.getTerm()) {
            logger.warn("[HandleLeaderTransfer] term changed, cur={} , request={}", memberState.currTerm(), request.getTerm());
            return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
        }

        return dLedgerRpcService.leadershipTransfer(takeLeadershipRequest).thenApply(response -> {
            synchronized (memberState) {
                // response不成功:清空受让人的id
                if (response.getCode() != DLedgerResponseCode.SUCCESS.getCode() ||
                    (memberState.currTerm() == request.getTerm() && memberState.getTransferee() != null)) {
                    logger.warn("leadershipTransfer failed, set transferee to null");
                    memberState.setTransferee(null); // 清空受让人id
                }
            }
            return response;
        });
    }

    /**
     * Leader转让时,当前节点是 受让人 .受让人处理接受Leader角色:
     *
     * 1、任期自增1
     * 2、使用新的任期,把角色改成 {@link Role#CANDIDATE}
     * 3、并标记可以理解自增term
     *
     * 理解: Leader转让给受让人时,其实就是'受让人'把term增加了1. 相当于受让人主动开启了一个新的任期,然后去投票选举.
     */
    public CompletableFuture<LeadershipTransferResponse> handleTakeLeadership(LeadershipTransferRequest request) throws Exception {
        logger.debug("handleTakeLeadership.request={}", request);
        synchronized (memberState) {
            if (memberState.currTerm() != request.getTerm()) {
                logger.warn("[BUG] [handleTakeLeadership] currTerm={} != request.term={}", memberState.currTerm(), request.getTerm());
                return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.INCONSISTENT_TERM.getCode()));
            }

            // term自增:开启下一个Leader任期
            long targetTerm = request.getTerm() + 1;
            memberState.setTermToTakeLeadership(targetTerm);
            CompletableFuture<LeadershipTransferResponse> response = new CompletableFuture<>();
            takeLeadershipTask.update(request, response);

            // Leader转让的逻辑,就是 受让人把term增加了1.相当于受让人主动开启了一个新的任期去投票选举
            changeRoleToCandidate(targetTerm);
            // 当前受让人变成 CANDIDATE 了,然后再把 needIncreaseTermImmediately 设置为true时,下一次 DLedgerLeaderElector#maintainAsCandidate 时,currTerm 才会自增的.
            needIncreaseTermImmediately = true;
            return response;
        }
    }

    /**
     * Leader角色转让
     * todo:受让人符合条件时.
     */
    private class TakeLeadershipTask {
        private LeadershipTransferRequest request;
        private CompletableFuture<LeadershipTransferResponse> responseFuture;

        public synchronized void update(LeadershipTransferRequest request, CompletableFuture<LeadershipTransferResponse> responseFuture) {
            this.request = request;
            this.responseFuture = responseFuture;
        }

        /**
         * 受让人经过校验去生成Leader转让请求的response
         *
         * @param term    当前任期
         * @param nowRole 变化后的新角色
         */
        public synchronized void check(long term, Role nowRole) {
            logger.trace("TakeLeadershipTask called, term={}, role={}", term, nowRole);
            // 如果当前不是'受让人',则这两个值都是空的,就会return了.
            if (memberState.getTermToTakeLeadership() == -1 || responseFuture == null) {
                return;
            }

            LeadershipTransferResponse response = null;
            if (term > memberState.getTermToTakeLeadership()) {
                response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.EXPIRED_TERM.getCode());
            }
            // 根据前面的逻辑,如果这两个值是相同时,说明在这中间还没有新的任期,此时,受让人才能接收Leader角色.
            else if (term == memberState.getTermToTakeLeadership()) {
                switch (nowRole) {
                    case LEADER:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.SUCCESS.getCode());
                        break;
                    case FOLLOWER:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.TAKE_LEADERSHIP_FAILED.getCode());
                        break;
                    default: // note:Leader转让的受让人在第一次check时,由于角色是 CANDIDATE ,所以这里直接返回了.
                            //   等受让人经过新一轮投票,确定了角色是 LEADER 还是 FOLLOWER,就知道该给 出让人 response什么了.
                        return;
                }
            } else {
                switch (nowRole) {
                    /*
                     * The node may receive heartbeat before term increase as a candidate,
                     * then it will be follower and term < TermToTakeLeadership
                     */
                    case FOLLOWER:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.TAKE_LEADERSHIP_FAILED.getCode());
                        break;
                    default:
                        response = new LeadershipTransferResponse().term(term).code(DLedgerResponseCode.INTERNAL_ERROR.getCode());
                }
            }

            responseFuture.complete(response);
            logger.info("TakeLeadershipTask finished. request={}, response={}, term={}, role={}", request, response, term, nowRole);
            memberState.setTermToTakeLeadership(-1);
            responseFuture = null;
            request = null;
        }
    }

    /**
     * 当DLedger角色变化的时候,会通知实现此接口的实现类.
     * RocketMQ使用DLedger模式时,会实现这个接口,用于感知角色变化.
     * 详见: https://gitee.com/anxiaole/rocketmq/blob/master/broker/src/main/java/org/apache/rocketmq/broker/dledger/DLedgerRoleChangeHandler.java
     */
    public interface RoleChangeHandler {
        void handle(long term, Role role);

        void startup();

        void shutdown();
    }

    /**
     * 角色维护线程
     *
     * @see Role
     */
    public class StateMaintainer extends ShutdownAbleThread {

        public StateMaintainer(String name, Logger logger) {
            super(name, logger);
        }

        @Override
        public void doWork() {
            try {
                // 如果启动Leader选举:
                if (DLedgerLeaderElector.this.dLedgerConfig.isEnableLeaderElector()) {
                    DLedgerLeaderElector.this.refreshIntervals(dLedgerConfig);
                    // 维持状态
                    DLedgerLeaderElector.this.maintainState();
                }
                sleep(10);
            } catch (Throwable t) {
                DLedgerLeaderElector.logger.error("Error in heartbeat", t);
            }
        }

    }
}
